package com.atguigu.dga230315.dolphinscheduler.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga230315.dolphinscheduler.bean.TDsTaskDefinition;
import com.atguigu.dga230315.dolphinscheduler.mapper.TDsTaskDefinitionMapper;
import com.atguigu.dga230315.dolphinscheduler.service.TDsTaskDefinitionService;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author zhangjunyi
 * @since 2023-07-19
 */
@Service
@DS("ds")
public class TDsTaskDefinitionServiceImpl extends ServiceImpl<TDsTaskDefinitionMapper, TDsTaskDefinition> implements TDsTaskDefinitionService {

    @Override
    public Map<String, TDsTaskDefinition> getTableDefinitionMapByDt(String assessDate, List<String> allAssessFullTableNameList) {
        //查询数据库 按照所包含的表 以及当前考评过的表查询
        List<TDsTaskDefinition> tDsTaskDefinitions = list(new QueryWrapper<TDsTaskDefinition>().in("name", allAssessFullTableNameList)
                .exists("select 1 from t_ds_task_instance ti where code=ti.task_code and date_format(ti.start_time,'%Y-%m-%d')='" + assessDate + "'")
        );

        //2.将数据转为Map集合  k：库名.表名  v：TDsTaskDefinition
        HashMap<String, TDsTaskDefinition> resutMap = new HashMap<>();
        for (TDsTaskDefinition tDsTaskDefinition : tDsTaskDefinitions) {
            //给sql这个属性赋值
            String taskParams = tDsTaskDefinition.getTaskParams();
            String sql=extractSQLFromShellTaskParams(taskParams);
            tDsTaskDefinition.setSql(sql);
            resutMap.put(tDsTaskDefinition.getName(), tDsTaskDefinition);
        }


        return resutMap;
    }

    private static String extractSQLFromShellTaskParams(String taskParams) {
        /**
         * taskParams：
         * {"resourceList":[],"localParams":[],"rawScript":"#!/bin/bash\n\nsql=\"\ninsert overwrite table gmall.ads_order_stats_by_cate\nselect * from gmall.ads_order_stats_by_cate\nunion\nselect\n    '。
         * 。
         * 。
         * 。
         * 。e\n)odr;\n\"\n\nhive -e \"$sql\"","dependence":{},"conditionResult":{"successNode":[],"failedNode":[]},"waitStartTimeout":{},"switchResult":{}}
         */

        //获取上面json字符中 rawScript对应的值，然后再把sql切出来
        JSONObject jsonObject = JSON.parseObject(taskParams);
        String rawScript = jsonObject.getString("rawScript");

        /**
         * rawScript:
         *"#!/bin/bash\n\nsql=\"\ninsert overwrite table gmall.ads_order_stats_by_cate\nselect * from gmall.ads_order_stats_by_cate\nunion\nselect\n    '${do_date}' dt,\n    recent_days,\n    category1_id,\n .
         * .
         * .
         * .
         * .r_nd lateral view explode(array(7,30)) tmp as recent_days\n        where dt='${do_date}'\n    )t1\n    group by recent_days,category1_id,category1_name,category2_id,category2_name,category3_id,category3_name\n)odr;\n\"\n\nhive -e \"$sql\""
         */

        //从上面获取到的字符串（shell脚本）切出sql

        //定义要切出来的字符串的开始下标和结束下标 然后根据下标获取到这个子串
        Integer beginIdx = -1;
        Integer endIdx = -1;

        //首先获取开始下标  sql语句开头有可能是 with...  有可能是insert.....

        //使用正则表达式去不区分大小写匹配 with  以及 insert
        //先匹配with
        Pattern compileWith = Pattern.compile("(?i)with");
        Matcher matcherWith = compileWith.matcher(rawScript);
        if (matcherWith.find()){
            beginIdx= rawScript.indexOf(matcherWith.group());
        }

        //如果没有with 就判断insert
        Pattern compileInsert = Pattern.compile("(?i)insert");
        Matcher matcherInsert = compileInsert.matcher(rawScript);
        if (beginIdx==-1&&matcherInsert.find()) { //sql语句没有with 有insert
            beginIdx= rawScript.indexOf(matcherInsert.group());
        }

        //最后获取结束下标 通常sql是以 ; 或者 "  能明确的是肯定得是双引号结尾  要求切出来的sql最后不能带; 因为sql解析器不识别这个语法
        //第二个参数的含义是 从指定下标位置开始找第一个参数的 这个字符下标
        String sql;

        int semiIdx = rawScript.indexOf(";", beginIdx);
        int quoteIdx = rawScript.indexOf("\"", beginIdx);
        if (semiIdx!=-1&&quoteIdx!=-1){
            //证明结尾既有双引号有用分号  去结束下标最小 也就是最靠近sql的
             sql = rawScript.substring(beginIdx,Math.min(semiIdx,quoteIdx));
        }else {
            //证明双引号 和 分号其中一个没有 去结束下标最大的，因为其中有一个是-1
            sql= rawScript.substring(beginIdx,Math.max(semiIdx,quoteIdx));
        }
        return sql;
    }

    public static void main(String[] args) {
        String paramJson = "{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"#!/bin/bash\\n\\nsql=\\\"\\n wITh ... insert overwrite table gmall.dwd_tool_coupon_used_inc partition(dt='${do_date}')\\nselect\\n    data.id,\\n    data.coupon_id,\\n    data.user_id,\\n    data.order_id,\\n    date_format(data.used_time,'yyyy-MM-dd') date_id,\\n    data.used_time\\nfrom gmall.ods_coupon_use_inc\\nwhere dt='${do_date}'\\nand type='update'\\nand array_contains(map_keys(old),'used_time');\\n\\\"\\n\\nhive -e \\\"$sql\\\"\",\"dependence\":{},\"conditionResult\":{\"successNode\":[],\"failedNode\":[]},\"waitStartTimeout\":{},\"switchResult\":{}}";
        String sql = extractSQLFromShellTaskParams(paramJson);
        System.out.println(sql);
    }
}
